package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Flink11_Transform_Process {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //2.从端口读取数据
        DataStreamSource<String> streamSource = env.socketTextStream("localhost", 9999);

        //TODO 3.使用process将数据转为JavaBean
        SingleOutputStreamOperator<WaterSensor> waterSensorDStream = streamSource.process(new ProcessFunction<String, WaterSensor>() {
            @Override
            public void processElement(String value, Context ctx, Collector<WaterSensor> out) throws Exception {
                String[] split = value.split(",");
                out.collect(new WaterSensor(split[0], Long.parseLong(split[1]), Integer.parseInt(split[2])));
            }
        });

        KeyedStream<WaterSensor, Tuple> keyedStream = waterSensorDStream.keyBy("id");

        //TODO 4.使用Process将数据的VC做累加 实现类似Sum的功能
        keyedStream.process(new KeyedProcessFunction<Tuple, WaterSensor, WaterSensor>() {
            //定义一个累加器计算累加结果
//            private Integer sumVc = 0;
            private HashMap<String, Integer> map = new HashMap<>();
            @Override
            public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                //判断map集合中是否有对应key的数据
                if (map.containsKey(value.getId())){
                    //key存在
                    //取出上一次累加过后的vc和
                    Integer lastVc = map.get(value.getId());

                    //与当前的vc进行累加
                    int currentVcSum = lastVc + value.getVc();

                    //将当前的vc累加过后的结果重新写入map集合
                    map.put(value.getId(), currentVcSum);

                    //将数据发送至下游
                    out.collect(new WaterSensor(value.getId(),value.getTs(),currentVcSum));
                }else {
                    //key不存在，证明这是第一条数据
                    //将这条数据存入map集合中
                    map.put(value.getId(), value.getVc());

                    out.collect(value);
                }
            }
        }).print();


        env.execute();
    }
}
